package bigdata.backend.monitor.warehouse.service.impl;

import bigdata.backend.config.DolphinSchedulerConfig;
import bigdata.backend.config.QualityMonitorConfig;
import bigdata.backend.monitor.warehouse.entity.QualityRule;
import bigdata.backend.monitor.warehouse.mapper.QualityRuleMapper;
import bigdata.backend.monitor.warehouse.service.QualityRuleService;
import bigdata.backend.task.dto.TaskNodeDto;
import bigdata.backend.task.service.DolphinSchedulerService;
import bigdata.backend.utils.SleepUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class QualityRuleServiceImpl extends ServiceImpl<QualityRuleMapper, QualityRule> implements QualityRuleService {
    @Autowired
    QualityRuleMapper qualityRuleMapper;

    @Autowired
    DolphinSchedulerConfig dolphinSchedulerConfig;

    @Autowired
    QualityMonitorConfig qualityMonitorConfig;

    @Autowired
    DolphinSchedulerService dolphinSchedulerService;

    // 读取有效条件 组装sql
    public List<QualityRule> getEnabledQualityRules() {
        LambdaQueryWrapper<QualityRule> qualityRuleLambdaQueryWrapper = new LambdaQueryWrapper<>();
        qualityRuleLambdaQueryWrapper.eq(QualityRule::getEnabled, true);
        qualityRuleLambdaQueryWrapper.eq(QualityRule::getDeleteFlag, '0');
        List<QualityRule> qualityRules = qualityRuleMapper.selectList(qualityRuleLambdaQueryWrapper);
        return qualityRules;
    }

    // 组装TaskNodeDto
    // 按照库.表进行分组 把同一个表的规则放到一起
    public List<TaskNodeDto> genTaskNodes(List<QualityRule> qualityRules) {
        Map<String, List<QualityRule>> qualityRuleByTableMap = new HashMap<>();
        for (QualityRule qualityRule : qualityRules) {
            String fullTableName = qualityRule.getFullTableName();
            // 如果 tableLabel 作为 key 已经存在于 Map 中，就返回已存在的 ArrayList
            // 如果 tableLabel 不存在，就执行 k -> new ArrayList<>() 创建一个新的 ArrayList，将其放入 Map 中，然后返回这个新的 ArrayList
            // 然后对返回的 ArrayList 执行 .add(qualityRule) 操作
            qualityRuleByTableMap.computeIfAbsent(fullTableName, arr -> new ArrayList<>()).add(qualityRule);
        }
        List<TaskNodeDto> taskNodes = new ArrayList<>();
        int n = 0;
        LocalDate today = LocalDate.now();
        String formatedDate = today.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
        String partition = qualityMonitorConfig.getPartitionPrefix() + formatedDate;
        for (Map.Entry<String, List<QualityRule>> entry : qualityRuleByTableMap.entrySet()) {
            String fullTableName = entry.getKey();
            String cteSql = "WITH base_table AS (SELECT * FROM " + fullTableName + ")\n";
            List<QualityRule> rules = entry.getValue();
            List<String> sqlList = new ArrayList<>();
            for (QualityRule rule : rules) {
                String sqlPart = buildSql(rule);
                sqlList.add(sqlPart);
            }
            String insertSql = "INSERT INTO " + qualityMonitorConfig.getQualityResultTable() + " PARTITION (" + partition + ") \n";
            if (n == 0) {
                insertSql = "INSERT OVERWRITE TABLE " + qualityMonitorConfig.getQualityResultTable() + " PARTITION (" + partition + ") \n";
            }
            String sql = insertSql + cteSql + String.join("\nUNION ALL\n", sqlList) + ";";
            TaskNodeDto taskNodeDto = new TaskNodeDto();
            String name = fullTableName.replace(".", "_");
            taskNodeDto.setName(name);
            taskNodeDto.setSqlScript(sql);
            taskNodeDto.setDatasourceId(dolphinSchedulerConfig.getDatasourceId());
            taskNodeDto.setOrder(n);
            taskNodes.add(taskNodeDto);
            n += 1;
        }
        return taskNodes;
    }

    @Override
    public void submitAllTask() throws Exception {
        String processName = "QualityRuleTaskByDay";
        List<QualityRule> qualityRules = getEnabledQualityRules();
        List<TaskNodeDto> taskNodeDtos = genTaskNodes(qualityRules);
        Long processCode = dolphinSchedulerService.recreateProcess(processName, taskNodeDtos, true);
        SleepUtil.sleep100ms();
        dolphinSchedulerService.releaseProcess(processCode, processName, "ONLINE");
        SleepUtil.sleep100ms();
        dolphinSchedulerService.runProcess(processCode);
        SleepUtil.sleep100ms();
        dolphinSchedulerService.createSchedule(processCode, qualityMonitorConfig.getCronExp());
        SleepUtil.sleep100ms();
        Integer scheduleId = dolphinSchedulerService.getScheduleId(processCode);
        SleepUtil.sleep100ms();
        dolphinSchedulerService.scheduleOnline(scheduleId);
    }

    private String buildSql(QualityRule rule) {

        String condition;
        if ("CUSTOM".equals(rule.getRuleType())) {
            condition = rule.getTriggerCondition();
        } else {
            condition = buildCondition(rule);
            condition = "NOT (" + condition + ")";
        }
        String outputJson = genJsonFunction(rule);
        String sql = "SELECT UUID() as id,\n" +
                "       '" + rule.getDatabaseName() + "' as database_name,\n" +
                "       '" + rule.getTableName() + "' as table_name,\n" +
                "       '" + rule.getRuleName() + "' as rule_name,\n" +
                "       '" + rule.getRuleCode() + "' as rule_code,\n" +
                "       " + rule.getPrimaryField() + " as primary_key,\n" +
                "       " + outputJson + " as output_fields,\n" +
                "       CURRENT_TIMESTAMP() as create_time\n" +
                "FROM base_table \n" +
                "WHERE " + condition;
        return sql;
    }

    private String genJsonFunction(QualityRule rule) {
        String outputFields = rule.getOutputFields();
        String[] fields = outputFields.split(",");
        StringBuilder jsonClause = new StringBuilder("json_object");
        jsonClause.append("(");
        for (int i = 0; i < fields.length; i++) {
            if (i > 0) jsonClause.append(", ");
            jsonClause.append("'").append(fields[i].trim()).append("', ").append(fields[i].trim());
        }
        jsonClause.append(")");
        return jsonClause.toString();
    }

    private String buildCondition(QualityRule rule) {
        // 宽松模式 SELECT '18366688607' REGEXP '^\\d+$';
        // 宽松模式 SELECT '183666886.07' REGEXP '^\\d+\\.\\d+$'
        // field IN ('', '')
        // SELECT char_length('123a');

        String simple = rule.getSimpleCondition();
        if (simple == null || !simple.contains(":")) {
            throw new IllegalArgumentException("Invalid simple_condition for rule: " + rule.getRuleCode());
        }

        String[] parts = simple.split(":", 2);
        String field = parts[0].trim();
        String spec = parts[1].trim();

        switch (rule.getRuleType()) {
            case "TYPE":
                return convertTypeCheck(field, spec);
            case "ENUM":
                return convertEnumCheck(field, spec);
            case "LENGTH":
                return convertLengthCheck(field, spec);
            case "ISNULL":
                return convertISNULLCheck(field);
            default:
                throw new IllegalArgumentException("Unsupported rule_type: " + rule.getRuleType());
        }
    }

    private String convertISNULLCheck(String field) {
        return "null_or_empty(" + field + ") = 0";
    }

    private String convertLengthCheck(String field, String lengthSpec) {
        int length = Integer.parseInt(lengthSpec);
        return "char_length(" + field + ") = " + length;
    }

    private String convertEnumCheck(String field, String enumSpec) {
        String[] values = enumSpec.split(",");
        StringBuilder inClause = new StringBuilder(field);
        inClause.append(" IN (");
        for (int i = 0; i < values.length; i++) {
            if (i > 0) inClause.append(",");
            inClause.append("'").append(values[i].trim()).append("'");
        }
        inClause.append(")");
        return inClause.toString();
    }

    private String convertTypeCheck(String field, String typeSpec) {
        switch (typeSpec.toLowerCase()) {
            case "int":
            case "integer":
                return field + " REGEXP '^\\\\d+$'";
            case "double":
            case "float":
                return field + " REGEXP '^\\\\d+\\\\.\\\\d+$'";
            default:
                throw new IllegalArgumentException("Unsupported type: " + typeSpec);
        }
    }


}
